package com.sh.data.engine.application.service.datadev;

import cn.hutool.core.convert.Convert;
import com.google.common.collect.Lists;
import com.sh.data.engine.application.model.dto.datadev.FlinkPointResponse;
import com.sh.data.engine.application.model.dto.datadev.FlinkTaskRecordResponse;
import com.sh.data.engine.application.model.request.datadev.RestartJobRequest;
import com.sh.data.engine.application.model.request.datadev.TaskRecordInfoRequest;
import com.sh.data.engine.common.util.FlinkUtil;
import com.sh.data.engine.common.util.FlinkUtil.FlinkCheckpointInfo;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.datadev.flink.model.domain.*;
import com.sh.data.engine.domain.datadev.flink.model.param.TaskRecordInfoParam;
import com.sh.data.engine.domain.datadev.flink.service.FlinkTaskV1Service;
import com.sh.data.engine.domain.util.ConvertUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @author: mengzheng.mhc
 * @date: 2024/8/9 18:20
 */
@Slf4j
@Service
public class FlinkRecordApplicationService {

    @Autowired
    private FlinkTaskV1Service flinkTaskV1Service;

    public PageResult<FlinkTaskRecordResponse> getTaskRecordPageInfo(TaskRecordInfoRequest taskRecordInfoRequest) {
        TaskRecordInfoParam taskRecordInfoParam =
            ConvertUtil.copyProperties(taskRecordInfoRequest, TaskRecordInfoParam.class);

        PageResult<FlinkLogRecordDomain> taskRecordPageInfo =
            flinkTaskV1Service.getTaskRecordPageInfo(taskRecordInfoParam);

        return ConvertUtil.convertPageInfo(taskRecordPageInfo, FlinkTaskRecordResponse.class);
    }

    public FlinkTaskRecordResponse getFlinkJobTaskDetailById(Long id) {
        FlinkJobDetailDomain flinkRecordDetailByrecordId =
            flinkTaskV1Service.getFlinkRecordDetailByrecordId(id);

        return Convert.convert(FlinkTaskRecordResponse.class, flinkRecordDetailByrecordId);
    }

    public FlinkLogDomain getTaskRecordDetailById(Long id) {
        return flinkTaskV1Service.getTaskRecordDetailById(id);
    }

    public boolean stopFlinkTaskById(Long id, Integer status) {
        return flinkTaskV1Service.stopFlinkJob(id, status);
    }

    public boolean retryJob(Long id) {
        return flinkTaskV1Service.retryJob(id);
    }

    public boolean restartJob(RestartJobRequest request) {
        RestartJobDomain convert = Convert.convert(RestartJobDomain.class, request);
        return flinkTaskV1Service.restartJob(convert);
    }

    public boolean continueJob(Long id, String point) {
        return flinkTaskV1Service.continueJob(id, point);
    }

    public List<FlinkPointResponse> getCheckPointsById(Long id) {
        List<FlinkUtil.FlinkCheckpointInfo> listCheckPoint = flinkTaskV1Service.getListCheckPoint(id);

        List<FlinkPointResponse> responseList = Lists.newArrayList();

        if (CollectionUtils.isNotEmpty(listCheckPoint)) {
            responseList =
                listCheckPoint.stream()
                    .map(
                        p -> {
                            FlinkPointResponse response = new FlinkPointResponse();
                            response.setTime(
                                DateFormatUtils.format(p.getCreateTime(), "yyyy-MM-dd HH:mm:ss"));
                            response.setPoint(p.getCheckpointPath());
                            return response;
                        })
                    .collect(Collectors.toList());
        }
        return responseList;
    }

    public FlinkPointResponse getLastSavepoint(Long id) {
        final FlinkCheckpointInfo lastSavepoint = flinkTaskV1Service.getLastSavepoint(id);
        FlinkPointResponse response = null;
        if (null != lastSavepoint) {
            response = new FlinkPointResponse();
            response.setTime(
                DateFormatUtils.format(lastSavepoint.getCreateTime(), "yyyy-MM-dd HH:mm:ss"));
            response.setPoint(lastSavepoint.getCheckpointPath());
        }
        return response;
    }

    public FlinkTaskSummaryDomain getSummary() {
        return flinkTaskV1Service.getSummary(ServletUtils.getProjectId());
    }
}
